package com.sjgs.gis.tilerpc;

import com.sjgs.common.utils.KryoUtils;
import com.sjgs.gis.domain.TileSet;
import com.sjgs.gis.domain.enumeration.TileSetType;
import com.sjgs.gis.fsrpc.HeaderServerInterceptor;
import com.sjgs.gis.kvtrans.KVTransServiceGrpc;
import com.sjgs.gis.kvtrans.BatchPutRequest;
import com.sjgs.gis.kvtrans.BatchPutResponse;
import com.sjgs.gis.kvtrans.MakeResponse;
import com.sjgs.gis.service.TileSetService;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import net.devh.boot.grpc.server.service.GrpcService;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.Resource;


@GrpcService(interceptors = {HeaderServerInterceptor.class})
public class TileGrpcService extends KVTransServiceGrpc.KVTransServiceImplBase {
    private static final Log logger = LogFactory.getLog(TileGrpcService.class.getName());

    @Value("${filesystem.grpc-chunk-size}")
    private Long grpcChunkSize;

    @Resource
    TileSetService tileSetService;

    /**
     */
    @Override
    public void makeTileSet(com.sjgs.gis.kvtrans.MakeRequest request,
                            io.grpc.stub.StreamObserver<com.sjgs.gis.kvtrans.MakeResponse> responseObserver) {
        TileSet tileSet = new TileSet();
        tileSet.setType(TileSetType.values()[request.getTileType().getNumber()]);
        if (request.getTileType().getNumber() == 0) {
            tileSet.setSchemaUrl(tileSet.getUid() + "/tms.json");
        }
        if (request.getTileType().getNumber() == 1) {
            tileSet.setSchemaUrl(tileSet.getUid() + "/layer.json");
        }
        if (request.getTileType().getNumber() == 2) {
            tileSet.setSchemaUrl(tileSet.getUid() + "/tileset.json");
            tileSet.setMetadataUrl(tileSet.getUid() + "/scenetree.json");
        }
        String context = Base64.encodeBase64String(KryoUtils.serializeObject(tileSet));
        MakeResponse response = MakeResponse.newBuilder()
                .setContext(context)
                .setMaxSize(grpcChunkSize)
                .build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }

    /**
     */
    @Override
    public io.grpc.stub.StreamObserver<com.sjgs.gis.kvtrans.BatchPutRequest> batchPut(
            io.grpc.stub.StreamObserver<com.sjgs.gis.kvtrans.BatchPutResponse> responseObserver) {
        final ServerCallStreamObserver<BatchPutResponse> serverCallStreamObserver =
                (ServerCallStreamObserver<BatchPutResponse>) responseObserver;
        // @2 禁止自动流控模式，开启手动流控
        serverCallStreamObserver.disableAutoInboundFlowControl();
        // @3 背压模式流控，当消费端有足够空间时将会回调OnReadyHandler
        // 默认空间大小为65536字节
        class OnReadyHandler implements Runnable {
            private boolean wasReady = false;

            @Override
            public void run() {
                if (serverCallStreamObserver.isReady() && !wasReady) {
                    wasReady = true;
                    logger.info("READY");
                    //即发信号通知发送端发送继续发消息
                    //并拉取客户端缓冲区消息
                    serverCallStreamObserver.request(1);
                }
            }
        }
        final OnReadyHandler onReadyHandler = new OnReadyHandler();
        serverCallStreamObserver.setOnReadyHandler(onReadyHandler);
        return new StreamObserver<BatchPutRequest>() {
            @Override
            public void onNext(BatchPutRequest request) {
                try {
                    TileSet tileSet = KryoUtils.deserializeObject(Base64.decodeBase64(request.getContext()), TileSet.class);
                    String uid = tileSet.getUid();
                    //batch write tile
                    BatchPutResponse response = BatchPutResponse.newBuilder().build();
                    responseObserver.onNext(response);
                    if (serverCallStreamObserver.isReady()) {
                        serverCallStreamObserver.request(1);
                    } else {
                        onReadyHandler.wasReady = false;
                    }
                } catch (Exception e) {
                    ;
                }
                logger.info("write" + request.getTilesCount());
            }

            @Override
            public void onError(Throwable t) {
                logger.error("put file exception");
                responseObserver.onCompleted();
            }

            @Override
            public void onCompleted() {
                responseObserver.onCompleted();
            }
        };
    }


}
